package ai.people.netmon.msgsearch.grpcservice;

import ai.people.core.grpc.lib.es.*;
import ai.people.core.grpc.utils.ProtoJsonUtil;
import ai.people.core.grpc.utils.StreamObserverDelegate;
import ai.people.netmon.framework.domain.searchmsg.request.AccountRequestToSpeak;
import ai.people.netmon.framework.domain.searchmsg.request.MessageRequest;
import ai.people.netmon.framework.domain.searchmsg.vo.AccountSpeakingLabelResponse;
import ai.people.netmon.framework.domain.searchmsg.vo.EsMessageDetailVO;
import ai.people.netmon.framework.model.response.Result;
import ai.people.netmon.msgsearch.service.EsSearchService;
import com.alibaba.fastjson.JSON;
import com.google.protobuf.Message;
import io.grpc.stub.StreamObserver;
import lombok.RequiredArgsConstructor;
import net.devh.boot.grpc.server.service.GrpcService;
import org.springframework.util.CollectionUtils;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;

/**
 * @author yuanqinglong
 * @date 2022/7/29 18:30
 */
@GrpcService
@RequiredArgsConstructor
public class SearchMessageGrpcService extends BaseSearchServiceGrpc.BaseSearchServiceImplBase {

    private final EsSearchService esSearchServiceImpl;


    private final ExecutorService defaultThreadPoolExecutor;

    /**
     * 基本搜索
     *
     * @param request          请求
     * @param responseObserver 响应观察者
     */
    @Override
    public void baseSearch(SearchRequest request, StreamObserver<SearchPageResponse> responseObserver) {
        MessageRequest messageRequest = ProtoJsonUtil.toJava(request, MessageRequest.class);
        StreamObserverDelegate.build(responseObserver).execute(() -> {
            Result search = esSearchServiceImpl.search(messageRequest, false);
            Object searchData = search.getData();
            Message message = ProtoJsonUtil.toProtoMessage(SearchPageResponse.newBuilder(), JSON.toJSONString(searchData));
            return (SearchPageResponse) message;
        });
    }


    /**
     * 搜索详情
     *
     * @param request          请求
     * @param responseObserver 响应观察者
     */
    @Override
    public void searchDetail(SearchDetailGrpcRequest request, StreamObserver<SearchDetailGrpcResponse> responseObserver) {
        StreamObserverDelegate.build(responseObserver).execute(() -> {
            try {
                EsMessageDetailVO esMessageDetailVO = esSearchServiceImpl.searchDetail(request.getId(), request.getPlatform());
                Message message = ProtoJsonUtil.toProtoMessage(SearchDetailGrpcResponse.newBuilder(), JSON.toJSONString(esMessageDetailVO));
                return (SearchDetailGrpcResponse) message;
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        });
    }


    /**
     * 批量搜索
     *
     * @param request          请求
     * @param responseObserver 响应观察者
     */
    @Override
    public void batchSearch(BatchSearchGrpcRequest request, StreamObserver<BatchSearchGrpcResponse> responseObserver) {
        List<PlatformFieldGrpcData> grpcDataList = request.getBatchSearchRequestList();
        StreamObserverDelegate.build(responseObserver).execute(() -> {
            BatchSearchGrpcResponse.Builder builder = BatchSearchGrpcResponse.newBuilder();
            if (!CollectionUtils.isEmpty(grpcDataList)) {
                List<CompletableFuture<EsMessageDetailVO>> futures = new ArrayList<>();
                // 异步执行查询
                for (PlatformFieldGrpcData platformFieldGrpcData : grpcDataList) {
                    CompletableFuture<EsMessageDetailVO> runAsync = CompletableFuture.supplyAsync(() -> {
                        String id = platformFieldGrpcData.getId();
                        String platform = platformFieldGrpcData.getPlatform();
                        try {
                            return esSearchServiceImpl.searchDetail(id, platform);
                        } catch (IOException e) {
                            throw new RuntimeException(e);
                        }

                    }, defaultThreadPoolExecutor);
                    futures.add(runAsync);
                }
                // 获取异步查询结果
                futures.forEach(esMessageDetailVOCompletableFuture -> {
                    try {
                        EsMessageDetailVO esMessageDetailVO = esMessageDetailVOCompletableFuture.get();
                        Message message = ProtoJsonUtil.toProtoMessage(SearchDetailGrpcResponse.newBuilder(), JSON.toJSONString(esMessageDetailVO));
                        builder.addList((SearchDetailGrpcResponse) message);
                    } catch (InterruptedException | ExecutionException e) {
                        throw new RuntimeException(e);
                    }
                });
                return builder.build();
            }
            return builder.build();
        });
    }

    /**
     * 查询账户警告言论
     *
     * @param request          请求
     * @param responseObserver 响应观察者
     */
    @Override
    public void queryAccountWarningSpeech(AccountWarningSpeechGrpcRequest request, StreamObserver<AccountWarningSpeechGrpcResponse> responseObserver) {
        StreamObserverDelegate.build(responseObserver).execute(() -> {
            AccountRequestToSpeak accountRequestToSpeak = ProtoJsonUtil.toJava(request, AccountRequestToSpeak.class);
            AccountWarningSpeechGrpcResponse.Builder newBuilder = AccountWarningSpeechGrpcResponse.newBuilder();
            try {
                List<AccountSpeakingLabelResponse> accountSpeakingLabelResponses = esSearchServiceImpl.queryAccountWarningRemarksCount(accountRequestToSpeak);
                for (AccountSpeakingLabelResponse accountSpeakingLabelResponse : accountSpeakingLabelResponses) {
                    Message message = ProtoJsonUtil.toProtoMessage(AccountWarningSpeechGrpcData.newBuilder(), JSON.toJSONString(accountSpeakingLabelResponse));
                    newBuilder.addList((AccountWarningSpeechGrpcData) message);
                }
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
            return newBuilder.build();
        });
    }
}
